package com.gzyj.flink.gps.map;

import com.gzyj.flink.dto.alarm.Kafka_VehicleNormalAlarm;
import com.gzyj.flink.dto.alarm.MonitorClientAlarm;
import com.gzyj.flink.gps.map.beidoualarm.BeidouPlatformAlarmProcess;
import com.gzyj.flink.gps.map.beidoualarm.BeidouZongduanDataProcess;
import com.gzyj.flink.gps.sink.gp.BaobiaoAlarmsummaryRecordSink;
import com.gzyj.flink.gps.sink.gp.BaobiaoAlarmsummarySink;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummary;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummaryRecord;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;


/**
 * 北斗报警数据处理
 */
public class BeidouAlarmProcess {

    private SingleOutputStreamOperator<Kafka_VehicleNormalAlarm> source;


    public BeidouAlarmProcess(SingleOutputStreamOperator<Kafka_VehicleNormalAlarm> source) {
        this.source = source;
    }


    public DataStream<BaobiaoAlarmsummary> process() {

        SingleOutputStreamOperator<Kafka_VehicleNormalAlarm> zongduanAlarmData = this.source.filter((alarm) -> {
            // 终端报警
            return alarm.getAnalyzeModel() == 0;
        });
        SingleOutputStreamOperator<Object> baobiaoAlarmsummaryStream = zongduanAlarmData.flatMap(new BeidouZongduanDataProcess());

        SingleOutputStreamOperator<BaobiaoAlarmsummary> saveBaobiaoAlarm = baobiaoAlarmsummaryStream.filter(e -> {
            return e instanceof BaobiaoAlarmsummary;
        }).map((d) -> {
            return (BaobiaoAlarmsummary) d;
        });


//        saveBaobiaoAlarm.addSink(new BaobiaoAlarmsummarySink());

        baobiaoAlarmsummaryStream.filter(e -> {
                    return e instanceof MonitorClientAlarm;
                })
                .map((d) -> {
                    return (MonitorClientAlarm) d;
                }).print();


        SingleOutputStreamOperator<Kafka_VehicleNormalAlarm> platformAlarm = this.source.filter((alarm) -> {
            return alarm.getAnalyzeModel() == 1;
        });


        SingleOutputStreamOperator<Object> beidouPlatformAlarmStream = platformAlarm.flatMap(new BeidouPlatformAlarmProcess());

        SingleOutputStreamOperator<BaobiaoAlarmsummary> platformAlarmSummaryStream= beidouPlatformAlarmStream.filter(e -> {
                    return e instanceof BaobiaoAlarmsummary;
                })
                .map((d) -> {
                    return (BaobiaoAlarmsummary) d;
                });



        beidouPlatformAlarmStream.filter(e -> {
                    return e instanceof BaobiaoAlarmsummaryRecord;
                })
                .map(d -> {
                    return (BaobiaoAlarmsummaryRecord) d;
                })
                .addSink(new BaobiaoAlarmsummaryRecordSink());

        beidouPlatformAlarmStream.filter(e -> {
                    return e instanceof MonitorClientAlarm;
                })
                .map(d -> {
                    return (MonitorClientAlarm) d;
                })
                .print();


        DataStream<BaobiaoAlarmsummary> resultStream=saveBaobiaoAlarm.union(platformAlarmSummaryStream);
        resultStream.addSink(new BaobiaoAlarmsummarySink());

        return resultStream;

    }


}
